--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit b36cd577f65bd8c3fc018fa71842d67e819fb4b5
Parents : bbc4f3b
Author : Ivan <ivan@quad4.io>
Signature : Invalid signer <e46112d44649266d71fe2193e00a4710>, author is <ivan@quad4.io>
Date : 2026-07-09T14:15:16-05:00
feat(i2p): implement I2P interface safety checks and configuration management
Changes
6 files changed, 1075 insertions(+), 15 deletions(-)
Diff
diff --git a/meshchatx/meshchat.py b/meshchatx/meshchat.py
index afc4ba5e..4cca7485 100644
--- a/meshchatx/meshchat.py
+++ b/meshchatx/meshchat.py
@@ -169,6 +169,7 @@ from meshchatx.src.backend.reticulum_config_guard import (
repair_unparseable_reticulum_config,
reticulum_config_has_required_sections,
)
+from meshchatx.src.backend import i2p_support
from meshchatx.src.backend.websocket_config_guard import (
sanitize_websocket_config_update,
websocket_type_requires_auth,
@@ -306,27 +307,45 @@ def _create_reticulum_instance(config_dir: str, loglevel: int | None = None):
Reticulum registers SIGINT/SIGTERM handlers in ``__init__``. Python only allows
``signal.signal`` on the main thread, so deferred network setup must skip that
registration when running in a background worker and install handlers later.
+
+ If the first init fails and the config still has an enabled I2P interface,
+ disable I2P and retry once so Android/desktop can recover without wiping
+ app data or the whole ``.reticulum`` tree.
"""
kwargs = {}
if loglevel is not None:
kwargs["loglevel"] = loglevel
- if threading.current_thread() is threading.main_thread():
- return RNS.Reticulum(config_dir, **kwargs)
+ def _construct():
+ if threading.current_thread() is threading.main_thread():
+ return RNS.Reticulum(config_dir, **kwargs)
+
+ real_signal = signal.signal
- real_signal = signal.signal
+ def _signal_allow_non_main(signum, handler):
+ try:
+ return real_signal(signum, handler)
+ except ValueError:
+ return signal.getsignal(signum)
- def _signal_allow_non_main(signum, handler):
+ signal.signal = _signal_allow_non_main
try:
- return real_signal(signum, handler)
- except ValueError:
- return signal.getsignal(signum)
+ return RNS.Reticulum(config_dir, **kwargs)
+ finally:
+ signal.signal = real_signal
- signal.signal = _signal_allow_non_main
try:
- return RNS.Reticulum(config_dir, **kwargs)
- finally:
- signal.signal = real_signal
+ return _construct()
+ except Exception as first_exc:
+ config_path = os.path.join(config_dir, "config")
+ if not i2p_support.disable_all_i2p_in_config(config_path):
+ raise
+ print(
+ "Reticulum init failed with I2P enabled; disabled I2P interfaces "
+ f"and retrying. Original error: {first_exc}",
+ flush=True,
+ )
+ return _construct()
def _install_reticulum_signal_handlers() -> bool:
@@ -1304,6 +1323,7 @@ class ReticulumMeshChat:
guard_rnode_interfaces_on_android(config_path)
guard_rnode_interfaces_on_desktop(config_path)
guard_invalid_rnode_txpower_in_config(config_path)
+ i2p_support.guard_i2p_interfaces_in_config(config_path)
def _set_startup_stage(self, stage: str, error: str | None = None) -> None:
self._startup_stage = stage
@@ -3790,6 +3810,13 @@ class ReticulumMeshChat:
try:
if hasattr(self, "reticulum") and self.reticulum:
self._sanitize_interfaces_section_names()
+ try:
+ i2p_support.repair_interfaces_dict(
+ self._get_interfaces_section(),
+ self._get_reticulum_section(),
+ )
+ except Exception as i2p_exc:
+ print(f"I2P config repair before write failed: {i2p_exc}")
self.reticulum.config.write()
self._verify_reticulum_config_reloadable()
return True
@@ -5743,6 +5770,14 @@ class ReticulumMeshChat:
status=404,
)
interface = interfaces[interface_name]
+ i2p_error = i2p_support.validate_i2p_enable(
+ interfaces,
+ self._get_reticulum_section(),
+ interface_name=interface_name,
+ )
+ if i2p_error is not None:
+ return web.json_response({"message": i2p_error}, status=422)
+
if "enabled" in interface:
interface["enabled"] = "true"
if "interface_enabled" in interface:
@@ -5905,6 +5940,16 @@ class ReticulumMeshChat:
status=422,
)
+ i2p_error = i2p_support.validate_i2p_add_or_update(
+ interfaces,
+ self._get_reticulum_section(),
+ interface_name=interface_name,
+ interface_type=interface_type,
+ updating_existing=bool(allow_overwriting_interface),
+ )
+ if i2p_error is not None:
+ return web.json_response({"message": i2p_error}, status=422)
+
# get existing interface details if available
interface_details = {}
if interface_name in interfaces:
@@ -6698,7 +6743,12 @@ class ReticulumMeshChat:
# merge new interface into existing interfaces
interfaces_before_write = self._get_interfaces_snapshot()
- interfaces[interface_name] = interface_details
+ if interface_type == "I2PInterface":
+ # I2P must be last: drop and reinsert so ConfigObj order is correct.
+ interfaces.pop(interface_name, None)
+ interfaces[interface_name] = interface_details
+ else:
+ interfaces[interface_name] = interface_details
# save config
if not self._write_reticulum_config(
rollback_interfaces=interfaces_before_write
@@ -6720,6 +6770,17 @@ class ReticulumMeshChat:
"message": "Interface has been saved",
},
)
+ if interface_type == "I2PInterface":
+ return web.json_response(
+ {
+ "message": (
+ "I2P interface has been added as the last interface. "
+ "Please restart MeshChat for these changes to take effect. "
+ "Do not add or reorder other interfaces afterward without "
+ "removing I2P first."
+ ),
+ },
+ )
return web.json_response(
{
"message": "Interface has been added. Please restart MeshChat for these changes to take effect.",
@@ -6795,6 +6856,12 @@ class ReticulumMeshChat:
# parse interfaces from config
interfaces = InterfaceConfigParser.parse(config)
+ # I2P must not be imported from files; hide it from the picker.
+ interfaces = [
+ iface
+ for iface in interfaces
+ if str(iface.get("type") or "").strip() != "I2PInterface"
+ ]
return web.json_response(
{
@@ -6859,6 +6926,15 @@ class ReticulumMeshChat:
iface_body = interface_config[interface_name]
iface_type = iface_body.get("type")
+ if iface_type == "I2PInterface" or i2p_support.is_i2p_interface(
+ iface_body
+ ):
+ return web.json_response(
+ {
+ "message": i2p_support.MSG_IMPORT_FORBIDDEN,
+ },
+ status=422,
+ )
if iface_type in ("RNodeInterface", "RNodeIPInterface"):
freq = iface_body.get("frequency")
if freq is not None and freq != "":
@@ -8622,6 +8698,10 @@ class ReticulumMeshChat:
# disable transport mode
reticulum_config = self._get_reticulum_section()
reticulum_config["enable_transport"] = False
+ i2p_support.disable_i2p_when_transport_off(
+ self._get_interfaces_section(),
+ reticulum_config,
+ )
if not self._write_reticulum_config():
return web.json_response(
{
@@ -8819,8 +8899,28 @@ class ReticulumMeshChat:
if not os.path.exists(config_dir):
os.makedirs(config_dir, exist_ok=True)
config_path = self._reticulum_config_file_path()
+ previous_interfaces = {}
+ if os.path.isfile(config_path):
+ try:
+ from RNS.vendor.configobj import ConfigObj
+
+ previous_interfaces = (
+ ConfigObj(config_path).get("interfaces") or {}
+ )
+ except Exception:
+ previous_interfaces = {}
+ i2p_raw_error = i2p_support.validate_raw_config_i2p_policy(
+ content,
+ previous_interfaces=previous_interfaces,
+ )
+ if i2p_raw_error is not None:
+ return web.json_response(
+ {"error": i2p_raw_error},
+ status=422,
+ )
with open(config_path, "w") as f:
f.write(content)
+ i2p_support.guard_i2p_interfaces_in_config(config_path)
return web.json_response(
{
"message": "Reticulum config saved",
@@ -11891,6 +11991,8 @@ class ReticulumMeshChat:
try:
session.start()
except Exception as e:
+ with contextlib.suppress(Exception):
+ manager.remove_session(session.session_id)
return web.json_response({"message": str(e)}, status=400)
return web.json_response(
{"session": session.to_dict(include_output_tail=True)}
diff --git a/meshchatx/src/backend/i2p_support.py b/meshchatx/src/backend/i2p_support.py
new file mode 100644
index 00000000..56c778f9
--- /dev/null
+++ b/meshchatx/src/backend/i2p_support.py
@@ -0,0 +1,390 @@
+# SPDX-License-Identifier: 0BSD
+
+"""I2PInterface safety helpers for MeshChatX.
+
+Reticulum's I2P interface is fragile: only one should exist, it must be the
+last interface in the config, and transport mode must already be enabled.
+Adding I2P via raw config/import or leaving it mid-list after later edits can
+brick startup (on Android that previously meant wiping the whole app).
+
+These helpers enforce the constraints on API writes and repair unsafe configs
+before Reticulum starts so identity/storage survive.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+_TRUE_STRINGS = ("true", "yes", "1", "on", "y")
+
+I2P_TYPE = "I2PInterface"
+
+MSG_TRANSPORT_REQUIRED = (
+ "Transport mode must be enabled before adding or enabling an I2P interface. "
+ "Enable it in Settings, then add I2P as the last interface."
+)
+MSG_ONLY_ONE = (
+ "Only one I2P interface is allowed. Remove or disable the existing I2P "
+ "interface before adding another."
+)
+MSG_IMPORT_FORBIDDEN = (
+ "I2P interfaces cannot be imported from a config file. Add I2P only through "
+ "the Add Interface page, with transport mode already enabled, as the last "
+ "interface."
+)
+MSG_RAW_FORBIDDEN = (
+ "I2P interfaces cannot be added or changed through the raw config editor. "
+ "Use the Add Interface page (transport must already be enabled), or delete "
+ "the I2P section from the file to recover."
+)
+MSG_MUST_BE_LAST = (
+ "The I2P interface must be the last interface in the Reticulum config. "
+ "Remove interfaces added after it, or delete I2P and re-add it last."
+)
+
+
+def is_i2p_interface(iface: object) -> bool:
+ if not isinstance(iface, dict):
+ return False
+ return str(iface.get("type") or "").strip() == I2P_TYPE
+
+
+def _parse_bool(value: object, *, default: bool = False) -> bool:
+ if value is None or value == "":
+ return default
+ if isinstance(value, bool):
+ return value
+ return str(value).strip().lower() in _TRUE_STRINGS
+
+
+def is_interface_enabled(iface: dict) -> bool:
+ for key in ("interface_enabled", "enabled"):
+ if key in iface:
+ return _parse_bool(iface.get(key), default=False)
+ return False
+
+
+def transport_enabled_in_section(reticulum_section: object) -> bool:
+ if not isinstance(reticulum_section, dict):
+ return False
+ return _parse_bool(reticulum_section.get("enable_transport"), default=False)
+
+
+def list_i2p_names(interfaces: object) -> list[str]:
+ if not isinstance(interfaces, dict):
+ return []
+ return [name for name, iface in interfaces.items() if is_i2p_interface(iface)]
+
+
+def _i2p_block_is_suffix(interfaces: dict) -> bool:
+ """True when every I2P entry is at the end (no non-I2P after the first I2P)."""
+ seen_i2p = False
+ for _name, iface in interfaces.items():
+ if is_i2p_interface(iface):
+ seen_i2p = True
+ elif seen_i2p:
+ return False
+ return True
+
+
+def i2p_is_last(interfaces: object) -> bool:
+ if not isinstance(interfaces, dict) or not interfaces:
+ return True
+ if not list_i2p_names(interfaces):
+ return True
+ return _i2p_block_is_suffix(interfaces)
+
+
+def validate_i2p_add_or_update(
+ interfaces: object,
+ reticulum_section: object,
+ *,
+ interface_name: str,
+ interface_type: str,
+ updating_existing: bool,
+) -> str | None:
+ """Return an error message when adding/updating I2P is unsafe, else None."""
+ if str(interface_type or "").strip() != I2P_TYPE:
+ return None
+ if not transport_enabled_in_section(reticulum_section):
+ return MSG_TRANSPORT_REQUIRED
+ if not isinstance(interfaces, dict):
+ interfaces = {}
+ existing = list_i2p_names(interfaces)
+ if updating_existing:
+ others = [n for n in existing if n != interface_name]
+ if others:
+ return MSG_ONLY_ONE
+ return None
+ if existing:
+ return MSG_ONLY_ONE
+ return None
+
+
+def validate_i2p_enable(
+ interfaces: object,
+ reticulum_section: object,
+ *,
+ interface_name: str,
+) -> str | None:
+ if not isinstance(interfaces, dict):
+ return None
+ target = interfaces.get(interface_name)
+ if not is_i2p_interface(target):
+ return None
+ if not transport_enabled_in_section(reticulum_section):
+ return MSG_TRANSPORT_REQUIRED
+ others_enabled = [
+ n
+ for n in list_i2p_names(interfaces)
+ if n != interface_name and is_interface_enabled(interfaces.get(n) or {})
+ ]
+ if others_enabled:
+ return MSG_ONLY_ONE
+ return None
+
+
+def validate_no_i2p_in_import(interface_config: dict) -> str | None:
+ for name, body in interface_config.items():
+ if is_i2p_interface(body):
+ return f'{MSG_IMPORT_FORBIDDEN} (rejected "{name}")'
+ return None
+
+
+def reorder_interfaces_i2p_last(interfaces: dict) -> bool:
+ """Move all I2P sections to the end. Returns True when order changed."""
+ if not isinstance(interfaces, dict) or not interfaces:
+ return False
+ if _i2p_block_is_suffix(interfaces):
+ return False
+
+ non_i2p: list[tuple[str, Any]] = []
+ i2p: list[tuple[str, Any]] = []
+ for name, iface in list(interfaces.items()):
+ if is_i2p_interface(iface):
+ i2p.append((name, iface))
+ else:
+ non_i2p.append((name, iface))
+ if not i2p:
+ return False
+
+ for name in list(interfaces.keys()):
+ del interfaces[name]
+ for name, iface in non_i2p + i2p:
+ interfaces[name] = iface
+ return True
+
+
+def enforce_single_enabled_i2p(interfaces: dict) -> bool:
+ """Disable all but the first enabled I2P interface. Returns True if changed."""
+ if not isinstance(interfaces, dict):
+ return False
+ modified = False
+ kept: str | None = None
+ for name, iface in interfaces.items():
+ if not is_i2p_interface(iface):
+ continue
+ if not is_interface_enabled(iface):
+ continue
+ if kept is None:
+ kept = name
+ continue
+ iface["interface_enabled"] = "false"
+ if "enabled" in iface:
+ iface["enabled"] = "false"
+ modified = True
+ logger.warning(
+ 'Disabled extra I2P interface "%s" (only one I2P interface is allowed; '
+ 'kept "%s")',
+ name,
+ kept,
+ )
+ return modified
+
+
+def disable_i2p_when_transport_off(
+ interfaces: dict,
+ reticulum_section: object,
+) -> bool:
+ if transport_enabled_in_section(reticulum_section):
+ return False
+ if not isinstance(interfaces, dict):
+ return False
+ modified = False
+ for name, iface in interfaces.items():
+ if not is_i2p_interface(iface):
+ continue
+ if not is_interface_enabled(iface):
+ continue
+ iface["interface_enabled"] = "false"
+ if "enabled" in iface:
+ iface["enabled"] = "false"
+ modified = True
+ logger.warning(
+ 'Disabled I2P interface "%s" because enable_transport is off',
+ name,
+ )
+ return modified
+
+
+def repair_interfaces_dict(
+ interfaces: dict,
+ reticulum_section: object,
+) -> bool:
+ """Apply all in-memory I2P safety repairs. Returns True when anything changed."""
+ if not isinstance(interfaces, dict):
+ return False
+ modified = False
+ if enforce_single_enabled_i2p(interfaces):
+ modified = True
+ if disable_i2p_when_transport_off(interfaces, reticulum_section):
+ modified = True
+ if reorder_interfaces_i2p_last(interfaces):
+ modified = True
+ logger.warning(
+ "Moved I2P interface(s) to the end of [interfaces] for safe startup",
+ )
+ return modified
+
+
+def existing_i2p_names_from_config_path(config_path: str) -> set[str]:
+ if not os.path.isfile(config_path):
+ return set()
+ try:
+ from RNS.vendor.configobj import ConfigObj
+
+ cfg = ConfigObj(config_path)
+ except Exception:
+ return set()
+ interfaces = cfg.get("interfaces")
+ return set(list_i2p_names(interfaces))
+
+
+def _iface_snapshot(iface: dict) -> dict:
+ out = {}
+ for key, value in iface.items():
+ if isinstance(value, dict):
+ continue
+ out[str(key)] = value
+ return out
+
+
+def validate_raw_config_i2p_policy(
+ content: str,
+ *,
+ previous_interfaces: dict | None = None,
+) -> str | None:
+ """Reject raw edits that add or alter I2P stanzas via the file editor.
+
+ Deleting an existing I2P section is allowed (recovery path). Adding I2P or
+ changing an existing I2P stanza through raw text is not.
+ """
+ try:
+ from RNS.vendor.configobj import ConfigObj
+ except Exception:
+ return None
+
+ try:
+ cfg = ConfigObj(content.splitlines())
+ except Exception:
+ return None
+
+ interfaces = cfg.get("interfaces")
+ if not isinstance(interfaces, dict):
+ return None
+
+ prev_ifaces = previous_interfaces if isinstance(previous_interfaces, dict) else {}
+ prev_names = set(list_i2p_names(prev_ifaces))
+ new_names = set(list_i2p_names(interfaces))
+
+ if new_names - prev_names:
+ return MSG_RAW_FORBIDDEN
+
+ if len(new_names) > 1:
+ return MSG_ONLY_ONE
+
+ if new_names and not _i2p_block_is_suffix(interfaces):
+ return MSG_MUST_BE_LAST
+
+ for name in new_names & prev_names:
+ prev_body = prev_ifaces.get(name) or {}
+ new_body = interfaces.get(name) or {}
+ if not isinstance(prev_body, dict) or not isinstance(new_body, dict):
+ return MSG_RAW_FORBIDDEN
+ if _iface_snapshot(prev_body) != _iface_snapshot(new_body):
+ return MSG_RAW_FORBIDDEN
+
+ return None
+
+
+def disable_all_i2p_in_config(config_path: str) -> bool:
+ """Disable every I2P interface in *config_path*. Returns True if changed."""
+ if not os.path.isfile(config_path):
+ return False
+ try:
+ from RNS.vendor.configobj import ConfigObj
+
+ cfg = ConfigObj(config_path)
+ except Exception:
+ return False
+ interfaces = cfg.get("interfaces")
+ if not isinstance(interfaces, dict):
+ return False
+ modified = False
+ for name, iface in interfaces.items():
+ if not is_i2p_interface(iface):
+ continue
+ if not is_interface_enabled(iface):
+ continue
+ iface["interface_enabled"] = "false"
+ if "enabled" in iface:
+ iface["enabled"] = "false"
+ modified = True
+ logger.warning(
+ 'Disabled I2P interface "%s" to recover from a Reticulum startup failure',
+ name,
+ )
+ if not modified:
+ return False
+ try:
+ cfg.write()
+ except Exception as exc:
+ logger.warning("Failed to disable I2P interfaces in config: %s", exc)
+ return False
+ return True
+
+
+def guard_i2p_interfaces_in_config(config_path: str) -> bool:
+ """Repair I2P entries in a Reticulum config file before startup.
+
+ Ensures at most one enabled I2P interface, disables I2P when transport is
+ off, and moves I2P sections to the end of [interfaces].
+ """
+ if not os.path.isfile(config_path):
+ return False
+ try:
+ from RNS.vendor.configobj import ConfigObj
+
+ cfg = ConfigObj(config_path)
+ except Exception as exc:
+ logger.warning("Could not open Reticulum config for I2P guard: %s", exc)
+ return False
+
+ interfaces = cfg.get("interfaces")
+ reticulum = cfg.get("reticulum")
+ if not isinstance(interfaces, dict):
+ return False
+
+ if not repair_interfaces_dict(interfaces, reticulum):
+ return False
+
+ try:
+ cfg.write()
+ except Exception as exc:
+ logger.warning("Failed to write I2P-guarded Reticulum config: %s", exc)
+ return False
+ return True
diff --git a/meshchatx/src/frontend/components/interfaces/AddInterfacePage.vue b/meshchatx/src/frontend/components/interfaces/AddInterfacePage.vue
index 7fa3ddea..0eb14c39 100644
--- a/meshchatx/src/frontend/components/interfaces/AddInterfacePage.vue
+++ b/meshchatx/src/frontend/components/interfaces/AddInterfacePage.vue
@@ -572,11 +572,31 @@
<!-- I2P Interface -->
<div v-if="newInterfaceType === 'I2PInterface'" class="space-y-4">
+ <div
+ class="bg-amber-50/80 dark:bg-amber-900/20 p-3 rounded-2xl border border-amber-200 dark:border-amber-800/40 text-xs text-amber-900 dark:text-amber-200 space-y-1"
+ >
+ <div class="font-semibold">
+ {{ $t("interfaces.i2p_requirements_title") }}
+ </div>
+ <p>{{ $t("interfaces.i2p_requirements_body") }}</p>
+ </div>
<div
class="bg-blue-50/50 dark:bg-blue-900/10 p-3 rounded-2xl border border-blue-100 dark:border-blue-900/20 text-xs text-blue-800 dark:text-blue-300"
>
- ⓘ To use the I2P interface, you must have an I2P router running on your
- system.
+ To use the I2P interface, you must have an I2P router running on your
+ system with SAM enabled.
+ </div>
+ <div
+ v-if="!transportEnabled"
+ class="bg-red-50/80 dark:bg-red-900/20 p-3 rounded-2xl border border-red-200 dark:border-red-800/40 text-xs text-red-800 dark:text-red-200"
+ >
+ {{ $t("interfaces.i2p_transport_required") }}
+ </div>
+ <div
+ v-else-if="hasExistingI2PInterface && !isEditingInterface"
+ class="bg-red-50/80 dark:bg-red-900/20 p-3 rounded-2xl border border-red-200 dark:border-red-800/40 text-xs text-red-800 dark:text-red-200"
+ >
+ {{ $t("interfaces.i2p_already_exists") }}
</div>
<div class="flex items-center gap-2">
<Toggle id="i2p-connectable" v-model="newInterfaceConnectable" />
@@ -1916,6 +1936,19 @@ export default {
newInterfaceBackboneListenPort: null,
newInterfaceBackboneListenIp: null,
newInterfaceBackboneListenDevice: null,
+ reticulumInstance: {
+ share_instance: true,
+ local_hops_delta: false,
+ respond_to_probes: false,
+ enable_remote_management: false,
+ shared_instance_type: "",
+ instance_name: "default",
+ rpc_key: null,
+ rpc_config_snippet: null,
+ is_connected_to_shared_instance: false,
+ enable_transport: false,
+ },
+ existingInterfaces: {},
sharedInterfaceSettings: {
mode: null,
@@ -2082,6 +2115,23 @@ export default {
}
return `${totalHz} Hz`;
},
+ transportEnabled() {
+ if (this.config && this.config.is_transport_enabled === true) {
+ return true;
+ }
+ return this.reticulumInstance.enable_transport === true;
+ },
+ hasExistingI2PInterface() {
+ return Object.values(this.existingInterfaces || {}).some(
+ (iface) => iface && iface.type === "I2PInterface"
+ );
+ },
+ canAddI2PInterface() {
+ if (this.isEditingInterface && this.newInterfaceType === "I2PInterface") {
+ return true;
+ }
+ return this.transportEnabled && !this.hasExistingI2PInterface;
+ },
},
watch: {
newInterfaceBandwidth: "updateRNodeCalculations",
@@ -2092,6 +2142,8 @@ export default {
},
mounted() {
this.getConfig();
+ this.loadReticulumInstance();
+ this.loadExistingInterfaces();
this.loadReticulumDiscoveryConfig();
this.loadComports();
this.loadHostKernelInterfaces();
@@ -2118,6 +2170,28 @@ export default {
console.log(e);
}
},
+ async loadReticulumInstance() {
+ try {
+ const response = await window.api.get(`/api/v1/reticulum/instance`);
+ if (response.data?.instance) {
+ this.reticulumInstance = {
+ ...this.reticulumInstance,
+ ...response.data.instance,
+ };
+ }
+ } catch (e) {
+ console.log(e);
+ }
+ },
+ async loadExistingInterfaces() {
+ try {
+ const response = await window.api.get(`/api/v1/reticulum/interfaces`);
+ this.existingInterfaces = response.data?.interfaces || {};
+ } catch (e) {
+ console.log(e);
+ this.existingInterfaces = {};
+ }
+ },
async updateConfig(config) {
try {
const response = await window.api.patch("/api/v1/config", config);
@@ -2791,6 +2865,10 @@ export default {
if (!config || !config.type || !config.name || this.isSaving) {
return;
}
+ if (config.type === "I2PInterface") {
+ ToastUtils.error(this.$t("interfaces.i2p_import_forbidden"));
+ return;
+ }
this.isSaving = true;
try {
const response = await window.api.post(
@@ -2840,6 +2918,15 @@ export default {
return;
}
+ if (this.newInterfaceType === "I2PInterface" && !this.canAddI2PInterface) {
+ ToastUtils.error(
+ !this.transportEnabled
+ ? this.$t("interfaces.i2p_transport_required")
+ : this.$t("interfaces.i2p_already_exists")
+ );
+ return;
+ }
+
if (this.newInterfaceType === "RNodeInterface" && this.newInterfaceRNodeUseBle) {
const raw = (this.newInterfaceRNodeBlePeer || "").trim();
const inner = raw.toLowerCase().startsWith("ble://") ? raw.slice(6).trim() : raw;
diff --git a/meshchatx/src/frontend/components/interfaces/ImportInterfacesModal.vue b/meshchatx/src/frontend/components/interfaces/ImportInterfacesModal.vue
index 56549208..1fef34c7 100644
--- a/meshchatx/src/frontend/components/interfaces/ImportInterfacesModal.vue
+++ b/meshchatx/src/frontend/components/interfaces/ImportInterfacesModal.vue
@@ -32,6 +32,7 @@
<ul class="list-disc list-inside">
<li>You can import interfaces from a ~/.reticulum/config file.</li>
<li>You can import interfaces from an exported interfaces file.</li>
+ <li>{{ $t("interfaces.i2p_import_forbidden") }}</li>
</ul>
</div>
</div>
diff --git a/tests/backend/test_i2p_support.py b/tests/backend/test_i2p_support.py
new file mode 100644
index 00000000..9db6b5e5
--- /dev/null
+++ b/tests/backend/test_i2p_support.py
@@ -0,0 +1,474 @@
+# SPDX-License-Identifier: 0BSD
+
+"""Regression tests for I2P interface safety and recovery."""
+
+from __future__ import annotations
+
+import contextlib
+import json
+import shutil
+import tempfile
+from unittest.mock import MagicMock, patch
+
+import pytest
+import RNS
+from RNS.vendor.configobj import ConfigObj
+
+from meshchatx.meshchat import ReticulumMeshChat
+from meshchatx.src.backend import i2p_support
+
+
+class ConfigDict(dict):
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.write_called = False
+
+ def write(self):
+ self.write_called = True
+ return True
+
+
+@pytest.fixture
+def temp_dir():
+ path = tempfile.mkdtemp()
+ try:
+ yield path
+ finally:
+ shutil.rmtree(path)
+
+
+def build_identity():
+ identity = MagicMock(spec=RNS.Identity)
+ identity.hash = b"test_hash_32_bytes_long_01234567"
+ identity.hexhash = identity.hash.hex()
+ identity.get_private_key.return_value = b"test_private_key"
+ return identity
+
+
+async def find_route_handler(app_instance, path, method):
+ for route in app_instance.get_routes():
+ if route.path == path and route.method == method:
+ return route.handler
+ return None
+
+
+@contextlib.asynccontextmanager
+async def make_app(temp_dir, config):
+ with (
+ patch("meshchatx.meshchat.generate_ssl_certificate"),
+ patch("RNS.Reticulum") as mock_rns,
+ patch("RNS.Transport"),
+ patch("LXMF.LXMRouter"),
+ ):
+ mock_reticulum = mock_rns.return_value
+ mock_reticulum.config = config
+ mock_reticulum.configpath = "/tmp/mock_config"
+ mock_reticulum.is_connected_to_shared_instance = False
+ mock_reticulum.transport_enabled.return_value = True
+
+ app_instance = ReticulumMeshChat(
+ identity=build_identity(),
+ storage_dir=temp_dir,
+ reticulum_config_dir=temp_dir,
+ )
+
+ yield app_instance
+
+
+def make_request(payload):
+ request = MagicMock()
+
+ async def _json():
+ return payload
+
+ request.json = _json
+ return request
+
+
+def test_reorder_interfaces_i2p_last():
+ interfaces = {
+ "A": {"type": "AutoInterface", "interface_enabled": "true"},
+ "I2P": {"type": "I2PInterface", "interface_enabled": "true"},
+ "B": {"type": "TCPClientInterface", "interface_enabled": "true"},
+ }
+ assert i2p_support.reorder_interfaces_i2p_last(interfaces) is True
+ assert list(interfaces.keys()) == ["A", "B", "I2P"]
+ assert i2p_support.i2p_is_last(interfaces) is True
+
+
+def test_enforce_single_enabled_i2p():
+ interfaces = {
+ "I2P1": {"type": "I2PInterface", "interface_enabled": "true"},
+ "I2P2": {"type": "I2PInterface", "interface_enabled": "true"},
+ }
+ assert i2p_support.enforce_single_enabled_i2p(interfaces) is True
+ assert i2p_support.is_interface_enabled(interfaces["I2P1"]) is True
+ assert i2p_support.is_interface_enabled(interfaces["I2P2"]) is False
+
+
+def test_disable_i2p_when_transport_off():
+ interfaces = {
+ "I2P": {"type": "I2PInterface", "interface_enabled": "true"},
+ }
+ assert (
+ i2p_support.disable_i2p_when_transport_off(
+ interfaces,
+ {"enable_transport": "False"},
+ )
+ is True
+ )
+ assert i2p_support.is_interface_enabled(interfaces["I2P"]) is False
+
+
+def test_guard_i2p_interfaces_in_config(tmp_path):
+ config_path = tmp_path / "config"
+ config_path.write_text(
+ """[reticulum]
+enable_transport = True
+
+[interfaces]
+[[A]]
+type = AutoInterface
+interface_enabled = true
+[[I2P]]
+type = I2PInterface
+interface_enabled = true
+peers = aaa.b32.i2p
+[[B]]
+type = TCPClientInterface
+interface_enabled = true
+[[I2P2]]
+type = I2PInterface
+interface_enabled = true
+peers = bbb.b32.i2p
+""",
+ encoding="utf-8",
+ )
+ assert i2p_support.guard_i2p_interfaces_in_config(str(config_path)) is True
+ cfg = ConfigObj(str(config_path))
+ names = list(cfg["interfaces"].keys())
+ assert names[-1] in ("I2P", "I2P2")
+ assert names[-2] in ("I2P", "I2P2")
+ enabled = [
+ n
+ for n, iface in cfg["interfaces"].items()
+ if iface.get("type") == "I2PInterface"
+ and i2p_support.is_interface_enabled(iface)
+ ]
+ assert len(enabled) == 1
+
+
+def test_guard_disables_i2p_without_transport(tmp_path):
+ config_path = tmp_path / "config"
+ config_path.write_text(
+ """[reticulum]
+enable_transport = False
+
+[interfaces]
+[[I2P]]
+type = I2PInterface
+interface_enabled = true
+peers = aaa.b32.i2p
+""",
+ encoding="utf-8",
+ )
+ assert i2p_support.guard_i2p_interfaces_in_config(str(config_path)) is True
+ cfg = ConfigObj(str(config_path))
+ assert i2p_support.is_interface_enabled(cfg["interfaces"]["I2P"]) is False
+
+
+def test_validate_raw_rejects_adding_i2p():
+ previous = {
+ "A": {"type": "AutoInterface", "interface_enabled": "true"},
+ }
+ content = """[reticulum]
+enable_transport = True
+[interfaces]
+[[A]]
+type = AutoInterface
+interface_enabled = true
+[[I2P]]
+type = I2PInterface
+interface_enabled = true
+peers = aaa.b32.i2p
+"""
+ err = i2p_support.validate_raw_config_i2p_policy(
+ content,
+ previous_interfaces=previous,
+ )
+ assert err == i2p_support.MSG_RAW_FORBIDDEN
+
+
+def test_validate_raw_allows_deleting_i2p():
+ previous = {
+ "A": {"type": "AutoInterface", "interface_enabled": "true"},
+ "I2P": {
+ "type": "I2PInterface",
+ "interface_enabled": "true",
+ "peers": ["aaa.b32.i2p"],
+ },
+ }
+ content = """[reticulum]
+enable_transport = True
+[interfaces]
+[[A]]
+type = AutoInterface
+interface_enabled = true
+"""
+ assert (
+ i2p_support.validate_raw_config_i2p_policy(
+ content,
+ previous_interfaces=previous,
+ )
+ is None
+ )
+
+
+def test_disable_all_i2p_in_config(tmp_path):
+ config_path = tmp_path / "config"
+ config_path.write_text(
+ """[reticulum]
+enable_transport = True
+[interfaces]
+[[I2P]]
+type = I2PInterface
+interface_enabled = true
+peers = aaa.b32.i2p
+""",
+ encoding="utf-8",
+ )
+ assert i2p_support.disable_all_i2p_in_config(str(config_path)) is True
+ cfg = ConfigObj(str(config_path))
+ assert i2p_support.is_interface_enabled(cfg["interfaces"]["I2P"]) is False
+
+
+@pytest.mark.asyncio
+async def test_add_i2p_requires_transport(temp_dir):
+ config = ConfigDict(
+ {
+ "reticulum": {"enable_transport": "False"},
+ "interfaces": {},
+ }
+ )
+ async with make_app(temp_dir, config) as app:
+ handler = await find_route_handler(
+ app,
+ "/api/v1/reticulum/interfaces/add",
+ "POST",
+ )
+ response = await handler(
+ make_request(
+ {
+ "name": "I2POut",
+ "type": "I2PInterface",
+ "peers": ["abcdef.b32.i2p"],
+ }
+ )
+ )
+ body = json.loads(response.body)
+ assert response.status == 422
+ assert "Transport" in body["message"]
+
+
+@pytest.mark.asyncio
+async def test_add_second_i2p_rejected(temp_dir):
+ config = ConfigDict(
+ {
+ "reticulum": {"enable_transport": "True"},
+ "interfaces": {
+ "ExistingI2P": {
+ "type": "I2PInterface",
+ "interface_enabled": "true",
+ "peers": ["aaa.b32.i2p"],
+ },
+ },
+ }
+ )
+ async with make_app(temp_dir, config) as app:
+ handler = await find_route_handler(
+ app,
+ "/api/v1/reticulum/interfaces/add",
+ "POST",
+ )
+ response = await handler(
+ make_request(
+ {
+ "name": "I2P2",
+ "type": "I2PInterface",
+ "peers": ["bbb.b32.i2p"],
+ }
+ )
+ )
+ body = json.loads(response.body)
+ assert response.status == 422
+ assert "Only one I2P" in body["message"]
+
+
+@pytest.mark.asyncio
+async def test_add_i2p_is_placed_last(temp_dir):
+ config = ConfigDict(
+ {
+ "reticulum": {"enable_transport": "True"},
+ "interfaces": {
+ "A": {"type": "AutoInterface", "interface_enabled": "true"},
+ },
+ }
+ )
+ async with make_app(temp_dir, config) as app:
+ handler = await find_route_handler(
+ app,
+ "/api/v1/reticulum/interfaces/add",
+ "POST",
+ )
+ response = await handler(
+ make_request(
+ {
+ "name": "I2POut",
+ "type": "I2PInterface",
+ "peers": ["abcdef.b32.i2p"],
+ }
+ )
+ )
+ body = json.loads(response.body)
+ assert response.status == 200, body
+ assert list(config["interfaces"].keys())[-1] == "I2POut"
+
+
+@pytest.mark.asyncio
+async def test_import_rejects_i2p(temp_dir):
+ config = ConfigDict(
+ {
+ "reticulum": {"enable_transport": "True"},
+ "interfaces": {},
+ }
+ )
+ async with make_app(temp_dir, config) as app:
+ handler = await find_route_handler(
+ app,
+ "/api/v1/reticulum/interfaces/import",
+ "POST",
+ )
+ response = await handler(
+ make_request(
+ {
+ "config": """[[I2POut]]
+type = I2PInterface
+peers = aaa.b32.i2p
+""",
+ "selected_interface_names": ["I2POut"],
+ }
+ )
+ )
+ body = json.loads(response.body)
+ assert response.status == 422
+ assert "cannot be imported" in body["message"]
+
+
+@pytest.mark.asyncio
+async def test_import_preview_hides_i2p(temp_dir):
+ config = ConfigDict({"reticulum": {}, "interfaces": {}})
+ async with make_app(temp_dir, config) as app:
+ handler = await find_route_handler(
+ app,
+ "/api/v1/reticulum/interfaces/import-preview",
+ "POST",
+ )
+ response = await handler(
+ make_request(
+ {
+ "config": """[[TCP]]
+type = TCPClientInterface
+target_host = 1.2.3.4
+target_port = 4242
+[[I2POut]]
+type = I2PInterface
+peers = aaa.b32.i2p
+""",
+ }
+ )
+ )
+ body = json.loads(response.body)
+ assert response.status == 200, body
+ names = [iface["name"] for iface in body["interfaces"]]
+ assert "TCP" in names
+ assert "I2POut" not in names
+
+
+@pytest.mark.asyncio
+async def test_raw_put_rejects_new_i2p(temp_dir):
+ config_path = f"{temp_dir}/config"
+ with open(config_path, "w", encoding="utf-8") as handle:
+ handle.write(
+ """[reticulum]
+enable_transport = True
+[interfaces]
+[[A]]
+type = AutoInterface
+interface_enabled = true
+"""
+ )
+ config = ConfigDict(
+ {
+ "reticulum": {"enable_transport": "True"},
+ "interfaces": {
+ "A": {"type": "AutoInterface", "interface_enabled": "true"},
+ },
+ }
+ )
+ async with make_app(temp_dir, config) as app:
+ handler = await find_route_handler(
+ app,
+ "/api/v1/reticulum/config/raw",
+ "PUT",
+ )
+ response = await handler(
+ make_request(
+ {
+ "content": """[reticulum]
+enable_transport = True
+[interfaces]
+[[A]]
+type = AutoInterface
+interface_enabled = true
+[[I2P]]
+type = I2PInterface
+interface_enabled = true
+peers = aaa.b32.i2p
+""",
+ }
+ )
+ )
+ body = json.loads(response.body)
+ assert response.status == 422
+ assert "raw config" in body["error"].lower() or "I2P" in body["error"]
+
+
+def test_create_reticulum_retries_after_disabling_i2p(tmp_path, monkeypatch):
+ config_path = tmp_path / "config"
+ config_path.write_text(
+ """[reticulum]
+enable_transport = True
+[interfaces]
+[[I2P]]
+type = I2PInterface
+interface_enabled = true
+peers = aaa.b32.i2p
+""",
+ encoding="utf-8",
+ )
+ calls = {"n": 0}
+
+ class FakeReticulum:
+ def __init__(self, *_args, **_kwargs):
+ calls["n"] += 1
+ if calls["n"] == 1:
+ raise RuntimeError("I2P brick")
+
+ monkeypatch.setattr("meshchatx.meshchat.RNS.Reticulum", FakeReticulum)
+ from meshchatx import meshchat as meshchat_mod
+
+ instance = meshchat_mod._create_reticulum_instance(str(tmp_path))
+ assert isinstance(instance, FakeReticulum)
+ assert calls["n"] == 2
+ cfg = ConfigObj(str(config_path))
+ assert i2p_support.is_interface_enabled(cfg["interfaces"]["I2P"]) is False
diff --git a/tests/backend/test_interface_options.py b/tests/backend/test_interface_options.py
index 0a3c9a74..3bf94a3b 100644
--- a/tests/backend/test_interface_options.py
+++ b/tests/backend/test_interface_options.py
@@ -715,7 +715,12 @@ async def test_ax25_kiss_persists_callsign_and_ssid(temp_dir):
@pytest.mark.asyncio
async def test_i2p_connectable_can_be_disabled(temp_dir):
- config = ConfigDict({"reticulum": {}, "interfaces": {}})
+ config = ConfigDict(
+ {
+ "reticulum": {"enable_transport": "True"},
+ "interfaces": {},
+ }
+ )
async with make_app(temp_dir, config) as handler:
payload = {
@@ -730,3 +735,4 @@ async def test_i2p_connectable_can_be_disabled(temp_dir):
saved = config["interfaces"]["I2POut"]
assert saved["connectable"] == "False"
assert saved["peers"] == ["abcdef.b32.i2p"]
+ assert list(config["interfaces"].keys())[-1] == "I2POut"
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────